package rocketmq

import (
	"context"
	"errors"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
)

type Publisher struct {
	rocketmq.Producer //生产者
	debug             bool
}

func (this *Publisher) Close() error {
	return this.Producer.Shutdown()
}

func (this *Publisher) Publish(topic string, msg []byte) error {
	if this.debug {
		Log.Writef("[发布][%s] : %s", topic, string(msg))
	}
	result, err := this.Producer.SendSync(context.Background(), primitive.NewMessage(topic, msg))
	if err != nil {
		return err
	}
	if result.Status != primitive.SendOK {
		switch result.Status {
		case primitive.SendFlushDiskTimeout:
			return errors.New("超时:Disk")
		case primitive.SendFlushSlaveTimeout:
			return errors.New("超时:Slave")
		case primitive.SendSlaveNotAvailable:
			return errors.New("未激活")
		case primitive.SendUnknownError:
			return errors.New("未知错误")
		default:
			return errors.New("发布失败")
		}
	}
	return nil
}

func NewPublisher(cfg *PublishConfig) (*Publisher, error) {
	c, err := rocketmq.NewProducer(
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{cfg.Endpoint})),
		producer.WithNamespace(cfg.InstanceId),
		producer.WithInstanceName(cfg.InstanceId),
		producer.WithRetry(2),
		producer.WithCredentials(primitive.Credentials{
			AccessKey: cfg.AccessKey,
			SecretKey: cfg.SecretKey,
		}),
	)
	if err != nil {
		return nil, err
	}
	return &Publisher{Producer: c, debug: cfg.Debug}, c.Start()
}
